﻿///////////////////////////////////////////////////////////////////////////////
//
//  This source is part of the  "Managed Task Framework.NET Runtime"  (MTF.NET Runtime).
//
//
//  Copyright (c) 2011, Multicore Consulting Limited (UK)
//  All rights reserved.
//  http://www.multicoreconsulting.co.uk
//
//  Redistribution and use in source and binary forms, with or without
//  modification, are permitted provided that the following conditions
//  are met:
//
//  -  Redistributions of source code must retain the above copyright
//     notice, this list of conditions and the following disclaimer.
//  -  Redistributions in binary form must reproduce the above copyright
//     notice, this list of conditions and the following disclaimer in
//     the documentation and/or other materials provided with the
//     distribution.
//  -  Neither the name of Multicore Consulting Limited nor the names of
//     its contributors may be used to endorse or promote products
//     derived from this software without specific prior written
//     permission.
//
//  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
//  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
//  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
//  FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
//  COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
//  INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
//  BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
//  LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
//  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
//  LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
//  ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
//  POSSIBILITY OF SUCH DAMAGE.
//
///////////////////////////////////////////////////////////////////////////////

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using MTFOrg.MTF.MTFCore.Messages;

namespace MTFOrg.MTF.MTFCore
{
	/// <summary>
	/// Class implementing the requester object.
	/// </summary>
	partial class Requester
	{
		/// <summary>
		/// Array of received events for each element.
		/// </summary>
		protected Event[] m_ReceivedEvents;

		/// <summary>
		/// Array of consumer slot sets associated with received events for each element.
		/// </summary>
		protected Dictionary<Event, int> m_ConsumerSlotIds;

        /// <summary>
        /// Semaphores for each element used to block until event is received.
        /// </summary>
        protected System.Threading.Semaphore[] m_EventSemaphores;

        /// <summary>
        /// Semaphores for each element used to block until a query is allowed.
        /// </summary>
        protected System.Threading.Semaphore[] m_CanQuerySemaphores;

        /// <summary>
        /// Number of active overlays.
        /// </summary>
        protected int[] m_OverlayCount;

        /// <summary>
        /// Maximum number of active overlays.
        /// </summary>
        protected int m_OverlayDepth;

        /// <summary>
        /// Number of threads requesting but blocked waiting for a slot to become free.
        /// </summary>
        protected int[] m_PendingCount;

        /// <summary>
        /// Whether a thread is blocked waiting for an event on a particular element.
        /// </summary>
        protected bool[] m_IsBlocked;

        /// <summary>
        /// Flag indicating whether the application is closing.
        /// </summary>
        private bool m_Closing = false;

        /// <summary>
        /// Delegate for an event received event.
        /// </summary>
        /// <param name="receivedEvent">The event that was requested</param>
        /// <param name="elementNumber">Element number of the requester</param>
        public delegate void EventReceivedDelegate(EventDictionary receivedEvent, int elementNumber);

        /// <summary>
        /// Holds the event received call-back.
        /// </summary>
        public event EventReceivedDelegate OnEventReceived;

		/// <summary>
		/// Called to initialize this task.
		/// </summary>
		public override void OnInitialize()
		{
			base.OnInitialize();

            // Store overlay depth
            m_OverlayDepth = OverlayDepth;

			// Set up received event arrays
			m_ReceivedEvents = new Event[m_ConsumerTask.NumElements];
			m_ConsumerSlotIds = new Dictionary<Event, int>();

            // Set up event semaphores
            m_EventSemaphores = new System.Threading.Semaphore[m_ConsumerTask.NumElements];
            for (int i = 0; i < m_ConsumerTask.NumElements; i++)
                m_EventSemaphores[i] = new System.Threading.Semaphore(0, 1);

            // Set up can-query semaphores
            m_CanQuerySemaphores = new System.Threading.Semaphore[m_ConsumerTask.NumElements];
            for (int i = 0; i < m_ConsumerTask.NumElements; i++)
                m_CanQuerySemaphores[i] = new System.Threading.Semaphore(1, 2);

            // Create blocking flags array
            m_IsBlocked = new bool[m_ConsumerTask.NumElements];

            // Create count arrays
            m_OverlayCount = new int[m_ConsumerTask.NumElements];
            m_PendingCount = new int[m_ConsumerTask.NumElements];

			// Hook up to the producer-consumer event notifications
			m_ConsumerTask.EventReceived += new ProducerConsumer.OnEvent(OnEventReceivedInternal);
            m_ConsumerTask.ReleasedReceived += new ProducerConsumer.OnReleased(OnReleasedReceived);

            CoreTaskManager.OnPreShutdown += new MTFOrg.MTF.MTFRuntime.ThreadPoolTaskManager.PreShutdownDelegate(CoreTaskManager_OnPreShutdown);
		}

        /// <summary>
        /// Called on pre-shutdown to release semaphores and allow closure.
        /// </summary>
        void CoreTaskManager_OnPreShutdown()
        {
            CoreTaskManager.OnPreShutdown -= new MTFOrg.MTF.MTFRuntime.ThreadPoolTaskManager.PreShutdownDelegate(CoreTaskManager_OnPreShutdown);

            m_Closing = true;
            foreach (System.Threading.Semaphore sem in m_EventSemaphores)
                sem.Release();
            foreach (System.Threading.Semaphore sem in m_CanQuerySemaphores)
                sem.Release();
        }

		/// <summary>
		/// Invoked when task execution commences.
		/// </summary>
		public override void OnBeginExecution()
		{
			base.OnBeginExecution();
		}

        /// <summary>
        /// Requests an event and waits until it arrives.
        /// </summary>
        /// <param name="elementNumber">Element number to request from</param>
        /// <returns>The requested event</returns>
        public object Request(int elementNumber)
        {
            return Request(elementNumber, true);
        }

		/// <summary>
		/// Requests an event and waits until it arrives.
		/// </summary>
		/// <param name="elementNumber">Element number to request from</param>
        /// <param name="blocking">True = block until event arrives</param>
		/// <returns>The requested event</returns>
		public object Request(int elementNumber, bool blocking)
		{
            // Block until a query is allowed
            m_CanQuerySemaphores[elementNumber].WaitOne();

            if (!m_Closing)
            {
                m_IsBlocked[elementNumber] = blocking;

                // Request event
                m_OverlayCount[elementNumber]++;
                m_ConsumerTask.SendQuery(0, Slots.NO_SLOT, elementNumber);

                if (blocking)
                {
                    // Block until event received
                    m_EventSemaphores[elementNumber].WaitOne();
                }
            }

            Event receivedEvent = null;

            if (blocking)
            {
                // Get received event
                receivedEvent = m_ReceivedEvents[elementNumber];

                // Allow next query to proceed
                releaseNextQuery(elementNumber);
            }

            return receivedEvent;
		}

		/// <summary>
		/// Called when the ProducerConsumer receives an event
		/// </summary>
		/// <param name="consumerLinkNum">Consumer link (absolute) on which the event arrived</param>
		/// <param name="consumerSlots">Consumer slot state information</param>
		/// <param name="producerSlots">Producer slot state information</param>
		/// <param name="receivedEvent">Event that was received</param>
		public void OnEventReceivedInternal(int consumerLinkNum, Slots slots, int consumerSlot, int producerSlot, Event eventReceived, ref object elementState, int elementNumber)
		{
			// Store event
			m_ReceivedEvents[elementNumber] = eventReceived;
			m_ConsumerSlotIds.Add(eventReceived, consumerSlot);

            if (m_IsBlocked[elementNumber])
            {
                // Unblock waiting thread
                m_EventSemaphores[elementNumber].Release();
            }
            else
            {
                // Asynchronously notify event received 
                OnEventReceived(eventReceived as EventDictionary, elementNumber);
                releaseNextQuery(elementNumber);
            }
		}

        /// <summary>
        /// Releases the next query.
        /// </summary>
        /// <param name="elementNumber">Element number</param>
        void releaseNextQuery(int elementNumber)
        {
            // Any space for another query?
            if (m_OverlayCount[elementNumber] < m_OverlayDepth)
            {
                // Signal the can-query semaphore to allow another query on this element number
                m_CanQuerySemaphores[elementNumber].Release();
            }
            else
            {
                // Release the requesting thread later
                m_PendingCount[elementNumber]++;
            }
        }

        /// <summary>
        /// Called when the producer notifies it has finished with an event.
        /// </summary>
        /// <param name="consumerLinkNum">Link that query was received on</param>
        /// <param name="elementState">State associated with the current element</param>
        /// <param name="elementNumber">Number of the current element</param>
        public void OnReleasedReceived(int consumerLinkNum, ref object elementState, int elementNumber)
        {
            // Finished with this overlay
            m_OverlayCount[elementNumber]--;

            // Any threads that couldn't get a slot earlier?
            if (m_PendingCount[elementNumber] > 0)
            {
                // Unblock thread that is waiting for a slot
                m_PendingCount[elementNumber]--;
                m_CanQuerySemaphores[elementNumber].Release();
            }
        }

        /// <summary>
        /// Releases the last requested event.
        /// </summary>
        /// <param name="elementNumber">Element number of event to release</param>
        public void Release(int elementNumber)
        {
            Release(elementNumber, m_ReceivedEvents[elementNumber]);

            // Clear out event ready for next time
            m_ReceivedEvents[elementNumber] = null;
        }

        /// <summary>
        /// Releases a previously requested event.
        /// </summary>
        /// <param name="elementNumber">Element number of event to release</param>
        /// <param name="eventToRelease">The event to be released</param>
        public void Release(int elementNumber, Event eventToRelease)
        {
            if (!m_Closing)
            {
                if (eventToRelease == null)
                    throw new Exception("Attempt to release event that has not been requested");

                // Dispose of the event and request again
                m_ConsumerTask.SendRelease(m_ConsumerSlotIds[eventToRelease], eventToRelease, false, 0, elementNumber, true);
                m_ConsumerSlotIds.Remove(eventToRelease);
            }
        }


        /// <summary>
        /// Extracts an object from an event using the specified key and index.
        /// </summary>
        /// <param name="key">Key to use to extract the object</param>
        /// <param name="index">Index of the object to extract</param>
        /// <param name="theEvent">Event to extract object from</param>
        /// <returns>Extracted object</returns>
        public object UnwrapObject(string key, int index, object theEvent)
        {
            EventDictionary eventDictionary = theEvent as EventDictionary;
            QueueEvent queueEvent = eventDictionary[key][index] as QueueEvent;
            return queueEvent.Item;
        }

        /// <summary>
        /// Stores an object in an event at the specified key and index.
        /// </summary>
        /// <param name="key">Key to store the object under</param>
        /// <param name="index">Index to store the object at</param>
        /// <param name="theEvent">Event to store the object in</param>
        /// <param name="obj">Object to store in the event</param>
        public void WrapObject(string key, int index, object theEvent, object obj)
        {
            EventDictionary eventDictionary = theEvent as EventDictionary;
            QueueEvent queueEvent = eventDictionary[key][index] as QueueEvent;
            queueEvent.Item = obj;
        }
    }
}